package com.learn.transform.partition;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

/**
 * @create: 2023-04-18 13:58
 * @author: Mr.Du
 * --------------
 * @notes: 全局分区，上游所有数据都发送到下游第一个算子实例
 **/
public class GlobalPartitioningDemo {
    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port", 8181);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        //并行度是1
        DataStreamSource<String> lines = env.socketTextStream("node1", 9999);

        SingleOutputStreamOperator<String> streamOperator = lines.map(new RichMapFunction<String, String>() {
            @Override
            public String map(String line) throws Exception {
                int indexOfThisSubtask = getRuntimeContext().getIndexOfThisSubtask();
                return line + ":" + indexOfThisSubtask;
            }
        }).setParallelism(2);

        DataStream<String> global = streamOperator.global();

        global.addSink(new RichSinkFunction<String>() {
            @Override
            public void invoke(String value, Context context) throws Exception {
                int indexOfThisSubtask = getRuntimeContext().getIndexOfThisSubtask();
                System.out.println(value + " -> " + indexOfThisSubtask);
            }
        });

        env.execute("");
    }
}
